package com.example.bicycle.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * 项目启动 监听主题
 *
 * @author Mr.Qu
 * @since 2020/11/18 0018
 */
@Slf4j
@Component
public class MQTTListener implements ApplicationListener<ContextRefreshedEvent> {

  @Value("${mqtt.username}")
  private String username;
  @Value("${mqtt.password}")
  private String password;
  private final MQTTConnect server;
  private final InitCallback initCallback;

  @Autowired
  public MQTTListener(MQTTConnect server, InitCallback initCallback) {
    this.server = server;
    this.initCallback = initCallback;
  }

  /**
   * 应用程序事件
   * 在监听器中直接订阅主题
   * @param contextRefreshedEvent 背景更新事件
   */
  @Override
  public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
    // 异步执行MQTT连接，避免阻塞应用启动
    connectToMqttWithRetry();
  }
  
  @Async
  private void connectToMqttWithRetry() {
    int maxRetries = 5;
    int retryDelay = 5000; // 5秒
    
    for (int i = 0; i < maxRetries; i++) {
      try {
        log.info("MQTT连接尝试 {}/{}", i + 1, maxRetries);
        
        // 等待EMQX服务完全启动
        if (i > 0) {
          Thread.sleep(retryDelay);
        }
        
        connectAndSubscribe();
        log.info("MQTT连接和订阅成功完成");
        return; // 成功后退出
        
      } catch (Exception e) {
        log.warn("MQTT连接尝试 {}/{} 失败: {}", i + 1, maxRetries, e.getMessage());
        
        if (i == maxRetries - 1) {
          log.error("MQTT连接失败，已达到最大重试次数", e);
        }
      }
    }
  }
  
  private void connectAndSubscribe() throws MqttException {
    log.info("正在连接MQTT服务器...");
    server.setMqttClient(username, password, initCallback);
    log.info("MQTT服务器连接成功");
    
    /*
    * 订阅数据主题
    * */
    server.sub("topic1");
    log.info("已订阅主题: topic1");
    
    /*
    * 订阅系统主题 - 分别订阅不同的事件类型
    * EMQX 5.x 版本的系统主题格式
    * */
    try {
        // 客户端连接事件
        server.sub("$SYS/brokers/+/clients/+/connected", 0);
        log.info("已订阅客户端连接事件主题: $SYS/brokers/+/clients/+/connected");
        
        // 客户端断开事件
        server.sub("$SYS/brokers/+/clients/+/disconnected", 0);
        log.info("已订阅客户端断开事件主题: $SYS/brokers/+/clients/+/disconnected");
        
        // 会话创建事件（可选）
        server.sub("$SYS/brokers/+/sessions/+/created", 0);
        log.info("已订阅会话创建事件主题: $SYS/brokers/+/sessions/+/created");
        
        // 会话终止事件（可选）
        server.sub("$SYS/brokers/+/sessions/+/terminated", 0);
        log.info("已订阅会话终止事件主题: $SYS/brokers/+/sessions/+/terminated");
        
        // 验证订阅是否成功
        if (server.isConnected()) {
            log.info("所有系统主题订阅成功完成");
        } else {
            log.error("MQTT客户端连接已断开");
        }
        
    } catch (MqttException e) {
        log.error("订阅系统主题失败: {}", e.getMessage());
        // 尝试订阅更通用的主题作为备用方案
        try {
            server.sub("$SYS/brokers/+/clients/#", 0);
            log.info("已订阅通用客户端事件主题: $SYS/brokers/+/clients/#");
        } catch (MqttException fallbackException) {
            log.error("备用订阅方案也失败: {}", fallbackException.getMessage());
            throw fallbackException;
        }
    }
  }
}